package com.edu.yx.app.APP_02_DimApp;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.edu.yx.app.APP_01_BaseApp.BaseApp;
import com.edu.yx.bean.TableProcess;
import com.edu.yx.common.Constant;
import com.edu.yx.util.FlinkSinkUtil;
import com.edu.yx.util.JdbcUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.Arrays;
import java.util.List;


/**
 *  1.etl数据流的全流程：
 *       mock -> mysql(开启binlog) -> maxwell读取binlog，获取数据装载语句，进入kafka的ods_db ->  kafka（zk）消费ods_db -> DimApp读取数据进行处理
 *
 *  2.tp配置流全流程：
 *      dim建表sql脚本 -> mysql(开启binlog) -> flink-cdc，直接获取到建表脚本的binlog日志  -> DimApp读取数据进行处理
 *
 *  3.etl数据流与tp配置流相结合确定phoenix的建表语句
 *
 *
 *  4.注意，dim的数据有首日导入（拿到历史数据），以及增量数据
 *      有一部分数据（只有首日导入），例如base_trademark、spu_info等，这些数据不会被jar包所改变
 *      如果不用maxwell--bootstrap去拿到全量数据，phoenix将有一部分表没数据
 */
public class DimApp extends BaseApp {

    public static void main(String[] args) {
        DimApp dimApp = new DimApp();

        //todo 这个初始化方法是将kafka的数据拿过来消费，而kafka数据是由maxwell提供
        dimApp.init(2000,2,"DimApp", Constant.TOPIC_ODS_DB);
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //todo 数据流etlStream
        // 1.数据为json格式，这个流的内容是各种dim表的内容，
        SingleOutputStreamOperator<JSONObject> etlStream = etl(stream);
//        etlStream.print();

        //todo 配置流tpStream
        //2.1 先执行sql建表语句，将dim的建表信息存到一个表中，同时mysql会产生执行建表语句的binlog日志
        //2.2 使用flink-cdc直接读取执行sql脚本产生的建表binlog日志，从binlog日志中读取建表的信息
        //2.3 将提取到的建表信息封装成一个pojo类并返回，返回配置流
        SingleOutputStreamOperator<TableProcess> tpStream = readTableProcess(env);


        //todo 3.这个自定义connect方法做了两件事
        // 3.1 在HBase建DIM表
        // 3.2 广播流保存DIM建表信息，etl数据流根据DIM建表信息，选择性的提取数据，并且封装成一个元祖往外抛
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dataTpStream = connect(etlStream, tpStream);


        //todo 4.根据dim建表信息内部的列，删除data没有与dim表具体的列相对应的数据，（data数据根据dim表的列有什么数据，保存什么数据）
        // 这一步把列以及数据都准备好
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream = filterNotNeedColumns(dataTpStream);

        //todo 5.将resultStream向phoenix写入数据
        writeToPhoenix(resultStream);
    }

    private void writeToPhoenix(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getPhoenixSink());
    }


    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> filterNotNeedColumns(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dataTpStream) {


        return dataTpStream
                .map(new MapFunction<Tuple2<JSONObject, TableProcess>, Tuple2<JSONObject, TableProcess>>() {
                    @Override
                    public Tuple2<JSONObject, TableProcess> map(  Tuple2<JSONObject, TableProcess> value  ) throws Exception {

                        //拿到所有数据，有些数据是多余的，表没有列去保存它
                        JSONObject data = value.f0;

                        //拿到所有列
                        List<String> columns = Arrays.asList(value.f1.getSinkcolumns().split(","));

                        //data是一个保存k-v的集合map,根据列的内容，如果列有某个字段，那么就保留data的数据，没有就删掉
                        //removeIf是选择性删除map元素
                        //op_type是后面自己加的列，在columns没有，记得不要删掉
                        data.keySet().removeIf(key -> !columns.contains(key) && !"op_type".equals(key));

                        return value;
                    }
                });

    }


    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect(SingleOutputStreamOperator<JSONObject> etlStream, SingleOutputStreamOperator<TableProcess> tpStream) {

        //todo 1.利用配置流（tpStream）创建表，在广播之前先建表
        //先根据tpStream读取到的表结构信息，在phoenix中创建相应的dim维度表,
        //利用rich版本的MapFunction创建连接和关闭连接
        tpStream.map(new RichMapFunction<TableProcess, TableProcess>() {


            private Connection conn;

            @Override
            public void open(Configuration parameters) throws Exception {
                //调用自己创建的工具类获取phoenix连接
                conn = JdbcUtil.getPhoenixConnection();
            }

            @Override
            public void close() throws Exception {
                JdbcUtil.closePhoenixConnection(conn);
            }

            //todo 真正的业务逻辑，根据读取到的建表信息进行建表
            @Override
            public TableProcess map(TableProcess value) throws Exception {

                //1.与mysql连接超过8h没响应会自动断开连接
                if (conn.isClosed()) {conn= JdbcUtil.getPhoenixConnection();}

                System.out.println("进行拼接");
                //2.拼建表语句
                StringBuilder sql = new StringBuilder();
                sql
                        .append("create table if not exists ")
                        .append(value.getSinktable()) //获取要输出到Phoenix的表
                        .append("(")
                        .append(value.getSinkcolumns().replaceAll("[^,]+","$0 varchar")) //利用正则表达式进行插入varchar完成字段的拼接
                        .append(",constraint pk primary key(")
                        .append(value.getSinkpk()== null ? "id" : value.getSinkpk()) //如果getSinkpk()为空，那么指定id作为主键，不为空就用getSinkpk()自己
                        .append("))") //一个是主键括号，一个是字段结束括号
                        .append(value.getSinkextend() == null ? "" : value.getSinkextend()); //value.getSinkextend()拼空字符串，不为空就拼自己，主要是用作后续的phoenix的盐表

                System.out.println("phoenix 正在拼接sql创建dim表" + sql );

                //3.获取sql语句，传进conn预处理
                PreparedStatement ps = conn.prepareStatement(sql.toString());

                //4.给sql中的占位符赋值(查增删改), ddl: 建表语句一般不会有占位符

                //5.执行
                ps.execute();

                //6.关ps
                ps.close();

                return value;
            }
        });


        //todo 2.将配置流（tpStream）做成带广播状态的广播流
        //2.1 创建一个状态，记录源表信息，方便与数据流etlStream中的json的源表信息进行匹配
        MapStateDescriptor<String, TableProcess> tpState = new MapStateDescriptor<>("tpState", String.class, TableProcess.class);
        //2.2 将配置流做成广播带广播状态的广播流（tpbcStream）
        BroadcastStream<TableProcess> tpbcStream = tpStream.broadcast(tpState);


        return etlStream
                .connect(tpbcStream)
                /**
                 *   @param <IN1> The input type of the non-broadcast side.
                 *   @param <IN2> The input type of the broadcast side.
                 *   @param <OUT> The output type of the operator.
                 */
                .process(new BroadcastProcessFunction<
                                                        JSONObject,  //<IN1> The input type of the non-broadcast side.
                                                        TableProcess, //<IN2> The input type of the broadcast side.
                                                        Tuple2<JSONObject,TableProcess>  //<OUT> The output type of the operator.
                                                    >() {

                    //处理数据流的数据
                    @Override
                    public void processElement(JSONObject value, //数据流中的内容
                                               ReadOnlyContext ctx,
                                               Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        //从状态tpState获取配置信息
                        ReadOnlyBroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpState);

                        /**
                         * 例子：
                         * {
                         *   "database": "gmall2022",
                         *   "table": "comment_info",
                         *   "type": "insert",
                         *   "ts": 1655176719,
                         *   "xid": 424843,
                         *   "xoffset": 14258,
                         *   "data": {。。。}
                         * }
                         */
                        //从数据流etlStream中拿到json内的key为table的内容
                        //etlStream中的数据都是ods_db，app运行指定的kafka的topic是ods_db
                        String key = value.getString("table");

                        //这个key是从ods_db拿出来的table表名，来自etlStream的表名
                        //state是所有dim表的建表信息，
                        //key只能拿到etlStream有的表名
                        TableProcess tableProcess = state.get(key);

                        //如果tableProcess为空，说明etlStream没有该表
                        //如果tableProcess不为空，说明etlStream有该表，进行数据提取，作为数据装载的数据，包装好抛出去
                        if (tableProcess!=null){

                            //data是dim表中填充列的数据,value是数据流etlStream的数据（JSON）
                            JSONObject data = value.getJSONObject("data");

                            //todo 额外加入op_type到data中，用于找到哪条数据是update的维度数据，方便redis第三方缓存判断
                            data.put("op_type",value.getString("type"));

                            //这里是将数据，以及对应的dim建表信息pojo类TableProcess对象一起扔出去
                            out.collect(Tuple2.of(data,tableProcess));

                        }


                    }

                    //处理广播流的数据
                    //广播流先对状态进行更新，让数据流可以读到状态，针对对应的表进行数据装载
                    //这个流专门负责提取配置流的信息，同步到状态上，数据流根据状态数据，动态获取表的信息
                    @Override
                    public void processBroadcastElement(TableProcess value, //数据由配置流进行供给
                                                        Context ctx,
                                                        Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        System.out.println();
                        System.out.println("-----------------广播流内的数据--------------------");
                        System.out.println(value.toString());
                        System.out.println("-----------------广播流内的数据--------------------");
                        System.out.println();

                        String key_sourceTable = value.getSourcetable();
                        BroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpState);
                        //将源表的表名，以及对应的整个TableProcess对象更新到状态，TableProcess包括了某个表的表名以及各种字段信息
                        state.put(key_sourceTable,value);
                    }
                });



    }


    private SingleOutputStreamOperator<TableProcess> readTableProcess(StreamExecutionEnvironment env) {
        //利用flink-cdc技术
        /**
         * CDC是Change Data Capture(变更数据获取)的简称。
         * 核心思想是，监测并捕获数据库的变动（包括数据或数据表的插入、更新以及删除等），
         * 将这些变更按发生的顺序完整记录下来，写入到消息中间件中以供其他服务进行订阅及消费。
         *
         * 1.这里创建的mySqlSource，作为flink-cdc直接读取了mysql的dim建表的binlog日志binlog日志
         */
        MySqlSource<String> mySqlSource = MySqlSource
                .<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("edu_config") //指定数据库
                .tableList("edu_config.table_process") //指定表命名
                .username("root")
                .password("123456")
                .deserializer(new JsonDebeziumDeserializationSchema()) //序列化为json
                .build();

         return env
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),"MySql Source")
                 //过滤一下after的value为null结果
                .filter(strJson->{
                    JSONObject jsonObject = JSON.parseObject(strJson);
                    return jsonObject.getString("after") != null;
                } )
                .map(strJson->{
                    JSONObject jsonObject = JSON.parseObject(strJson);
                    //将json中after的数据拿出来，上述的json中，有个key为after
                    // json封装成一个pojo对象，TableProcess
                    System.out.println();
                    System.out.println("-------------cdc读取到的内容------------------");
                    System.out.println(strJson);
                    System.out.println("-------------cdc读取到的内容------------------");
                    System.out.println();


                    return jsonObject.getObject("after", TableProcess.class);
                });

    }


    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {

         return stream

                 //这个json是maxwell采集到kafka的数据，首日以及每日采集数据没有区别，都是实时获取
                .filter(stringJson -> {

                    try {
                        /**
                         * 1、bootstrap使用
                         *  Maxwell允许您将数据“引导”到流中。这将执行 select * from table和将结果输出到您的流中，从而允许您从头开始播放流来重新创建整个数据集。

                         * 您可以使用该maxwell-bootstrap实用程序从命令行开始boostrap操作
                         *
                         */
                        //
                        JSONObject jsonObject = JSON.parseObject(stringJson.replace("bootstrap-", ""));

                        return "edu".equals(jsonObject.getString("database"))
                                &&
                                ("insert".equals(jsonObject.getString("type")) || "update".equals(jsonObject.getString("type")))
                                &&
                                jsonObject.getString("data") != null
                                &&
                                jsonObject.getString("data").length() > 2;
                    } catch (Exception e) {

                        //如果数据不是json，捕获异常
                        System.out.println("检测到非json数据: " + stringJson);
                        e.printStackTrace();
                        return false;
                    }


                })
                .map(json -> JSON.parseObject(json));

    }

}
